package com.zzw.demo.util;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;


import java.util.List;

public class CastDebeziumDeserializationSchema implements DebeziumDeserializationSchema {


    @Override
    public void deserialize(SourceRecord sourceRecord, Collector collector) throws Exception {
        //获取表名和数据库名

        String[] split = sourceRecord.topic().split("[.]");
        //数据库名
        String database = split[1];
        //表名
        String table = split[2];

        Struct struct =(Struct) sourceRecord.value();
        //添加修改前的  before
        Struct before = struct.getStruct("before");
        JSONObject beforeObject = new JSONObject();
        if (before != null){
            List<Field> beforeList = before.schema().fields();
            for (Field field : beforeList) {
                Object value = before.get(field);
                beforeObject.put(field.name(),value);
            }
        }


        //添加修改后的  after
        Struct after = struct.getStruct("after");
        JSONObject afterObject = new JSONObject();
        if (after != null){
            List<Field> fields = after.schema().fields();
            for (Field field : fields) {
                Object value = after.get(field);
                afterObject.put(field.name(),value);
            }
        }


        //操作类型
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
        if ("create".equals(type)){
            type = "insert";
        }
        //写入JSON
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("database",database);
        jsonObject.put("table",table);
        jsonObject.put("type",type);
        jsonObject.put("after",afterObject);
        jsonObject.put("before",beforeObject);

        //输出
        collector.collect(jsonObject.toJSONString());

    }

    @Override
    public TypeInformation getProducedType() {
       return TypeInformation.of(new TypeHint<String>() {
       });
    }
}
